package com.gitee.kamismile.gatewayAgent.client.bootstrap;

import com.gitee.kamismile.gatewayAgent.client.handler.GateWayClientInitializer;
import com.gitee.kamismile.gatewayAgent.client.service.IGateWayClientHelper;
import com.gitee.kamismile.stone.commmon.exception.BusinessException;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Component;

import java.net.SocketAddress;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by lidong on 2017/2/8.
 */
@Component
public class GateWayClient {
    private final org.slf4j.Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    GatewayAgentClientConstant gatewayAgentClientConstant;

    @Autowired
    IGateWayClientHelper gateWayClientHelper;

    @Autowired
    GateWayClientInitializer gateWayClientInitializer;

    private final Lock connectLock = new ReentrantLock();

    private Bootstrap bootstrap;
    private EventLoopGroup workerGroup;

    private Channel channel;

    private static final ScheduledThreadPoolExecutor reconnectExecutorService = new ScheduledThreadPoolExecutor(2, new CustomizableThreadFactory("GatewayAgentClientReconnectTimer"));

    private volatile ScheduledFuture<?> reconnectExecutorFuture = null;

    private final AtomicInteger reconnect_count = new AtomicInteger(0);

    //重连的error日志是否已经被调用过.
    private final AtomicBoolean reconnect_error_log_flag = new AtomicBoolean(false);

    //重连warning的间隔.(waring多少次之后，warning一次) //for test
    private final int reconnect_warning_period = 2;

    private final long shutdown_timeout = 1000 * 60 * 15;

    private long lastConnectedTime = System.currentTimeMillis();

    public void start() throws Exception {
        try {
            doOpen();
            connect();
        } catch (Exception e) {
//            close();
        }
    }


    public void doOpen() throws Exception {
        String os = System.getProperty("os.name");
        bootstrap = new Bootstrap();
        if (os.startsWith("win") || os.startsWith("Win")) {
            workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);
            bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(gateWayClientInitializer);
        } else {
            workerGroup = new EpollEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);
            bootstrap.group(workerGroup)
                    .channel(EpollSocketChannel.class)
                    .handler(gateWayClientInitializer);
        }
    }


    protected void connect() throws Exception {
        connectLock.lock();
        try {
            if (isConnected()) {
                return;
            }
            initConnectStatusCheckCommand();
            doConnect();
            if (!isConnected()) {
                throw new BusinessException(-1, "Failed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " "
                       );
            } else {
                logger.info("Successed connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() +
                         ", channel is " + this.getChannel());
            }
            reconnect_count.set(0);
            reconnect_error_log_flag.set(false);
        } finally {
            connectLock.unlock();
        }
    }

    private synchronized void initConnectStatusCheckCommand() {
        //reconnect=false to close reconnect
        if (reconnectExecutorFuture == null || reconnectExecutorFuture.isCancelled()) {
            Runnable connectStatusCheckCommand = new Runnable() {
                public void run() {
                    try {
                        if (!isConnected()) {
                            connect();
                        } else {
                            lastConnectedTime = System.currentTimeMillis();
                        }
                    } catch (Throwable e) {
                        String errorMsg = "client reconnect to " + getRemoteAddress() + " find error: ";
                        // wait registry sync provider list
                        if (System.currentTimeMillis() - lastConnectedTime > shutdown_timeout) {
                            if (!reconnect_error_log_flag.get()) {
                                reconnect_error_log_flag.set(true);
                                logger.warn(errorMsg);
                                return;
                            }
                        }
                        if (reconnect_count.getAndIncrement() % reconnect_warning_period == 0) {
                            logger.warn(errorMsg);
                        }
                    }
                }
            };
            reconnectExecutorFuture = reconnectExecutorService.scheduleWithFixedDelay(connectStatusCheckCommand, 2, 3, TimeUnit.SECONDS);
        }
    }


    protected void doConnect() throws Exception {
        ChannelFuture future = bootstrap.connect(gatewayAgentClientConstant.getInetHost(), gatewayAgentClientConstant.getPort());
        boolean ret = future.awaitUninterruptibly(3, TimeUnit.SECONDS);
        if (ret && future.isSuccess()) {
            Channel newChannel = future.channel();

            try {
                Channel oldChannel = this.channel;
                if (oldChannel != null) {
                    oldChannel.close();
                }
            } finally {
                this.channel = newChannel;
                if (!isConnected()) {
                    future.cancel(true);
                } else {
                    gateWayClientHelper.afterConnect();
                }
            }
        }
    }


    public boolean isConnected() {
        Channel channel = getChannel();
        if (channel == null)
            return false;
        return channel.isActive();
    }

    public Channel getChannel() {
        return channel;
    }


    public SocketAddress getRemoteAddress() {
        Channel channel = getChannel();
        if (channel == null)
            return null;
        return channel.remoteAddress();
    }

    public void close() throws Exception {
        if (workerGroup != null) {
            workerGroup.shutdownGracefully().sync();
        }
        disconnect();
    }

    public void disconnect() {
        connectLock.lock();
        try {
            destroyConnectStatusCheckCommand();
            Channel channel = getChannel();
            if (channel != null) {
                channel.close();
            }
        } finally {
            connectLock.unlock();
        }
    }

    private synchronized void destroyConnectStatusCheckCommand() {
        try {
            if (reconnectExecutorFuture != null && !reconnectExecutorFuture.isDone()) {
                reconnectExecutorFuture.cancel(true);
                reconnectExecutorService.purge();
            }
        } catch (Throwable e) {
            logger.warn(e.getMessage());
        }
    }


//    public void sendMessage(String cmdInfo, String data, Long seq, String router) {
//        if (ValueUtils.isNotNull(channel)) {
//            GateWayClientHandler handler = channel.pipeline().get(GateWayClientHandler.class);
//            handler.sendMessage(cmdInfo, data, seq, router);
//        }
//    }


}
